package com.hyk.learning.registry.zk.util;

import com.hyk.learning.common.enums.RpcConfigEnum;
import com.hyk.learning.common.utils.PropertiesFileUtil;
import com.hyk.learning.remoting.cache.ServiceCache;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @className: CuratorUtils
 * @author: hyk199710
 * @description: TODO
 * @date: 2024/5/8 22:53
 */
@Slf4j
public class CuratorUtils {

    private static final int BASE_SLEEP_TIME = 1000;
    private static final int MAX_RETRIES = 3;
    public static final String ZK_REGISTER_ROOT_PATH = "/my-rpc";
    //添加本地缓存
    private static final ServiceCache cache = new ServiceCache();
    private static final Set<String> REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet();
    private static CuratorFramework zkClient;
    private static final String DEFAULT_ZOOKEEPER_ADDRESS = "127.0.0.1:2181";

    private CuratorUtils() {
    }

    /**
     * @param zkClient:
     * @param path:
     * @return void
     * @author 10091
     * @description create presistent node
     * @date 2024/5/8 23:00
     */
    public static void createPersistentNode(CuratorFramework zkClient, String path) {
        try {
            if (REGISTERED_PATH_SET.contains(path) || zkClient.checkExists().forPath(path) != null) {
                log.info("The node already exists. The node is: [{}]", path);
            } else {
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                log.info("The node was created successfully. The node is: [{}]", path);
            }
        } catch (Exception e) {
            log.error("Create persistent node for path [{}] fail.", path);
        }
    }

    public static List<String> getChildrenNodes(CuratorFramework zkClient, String rpcServiceName) {
        List<String> result = cache.getServiceFromCache(rpcServiceName);
        if (null != null && !result.isEmpty()) {
            return result;
        }

        String servicePath = ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName;
        try {
            result = zkClient.getChildren().forPath(servicePath);
            result.stream().forEach(address -> {cache.addServiceToCache(rpcServiceName, address);});
        } catch (Exception e) {
            log.error("Get children nodes for path [{}] fail.", servicePath);
        }
        return result;
    }

    public static void registerWathcer() throws Exception {
        CuratorCache curatorCache = CuratorCache.build(getZkClient(), "/");
        curatorCache.listenable().addListener(new CuratorCacheListener() {
            @Override
            public void event(Type type, ChildData childData, ChildData childData1) {
                switch(type.name()){
                    case "NODE_CREATED":
                        String path = childData1.getPath();
                        String[] pathList = path.split("/");
                        if(pathList.length <= 2) break;
                        else{
                            String serviceName = pathList[1];
                            String address = pathList[2];
                            cache.addServiceToCache(serviceName, address);
                        }
                        break;
                    case "NODE_CHANGED":
                        log.info("修改前的数据：{}", childData.getData());
                        log.info("修改后的数据：{}", childData1.getData());
                        break;
                    case "NODE_DELETED":
                        String path_del = childData1.getPath();
                        String[] delList = path_del.split("/");
                        if(delList.length <= 2) break;
                        else{
                            String serviceName = delList[1];
                            String address = delList[2];
                            cache.delete(serviceName, address);
                        }
                        break;
                    default:
                        break;
                }
            }
        });
        curatorCache.start();
    }

    public static void clearRegistry(CuratorFramework zkClient, InetSocketAddress inetSocketAddress) {
        REGISTERED_PATH_SET.stream().parallel().forEach(p -> {
            try {
                if (p.endsWith(inetSocketAddress.toString())) {
                    zkClient.delete().forPath(p);
                }
            } catch (Exception e) {
                log.error("clear registry for path [{}] fail", p);
            }
        });
        log.info("All registered services on the server are cleared:[{}]", REGISTERED_PATH_SET);
    }

    public static CuratorFramework getZkClient() {
        Properties properties = PropertiesFileUtil.readPropertiesFile(RpcConfigEnum.RPC_CONFIG_ENUM.getPropertyValue());
        String zookeeperAddress = properties != null
                && properties.getProperty(RpcConfigEnum.ZK_ADDRESS.getPropertyValue())
                != null ? properties.getProperty(RpcConfigEnum.ZK_ADDRESS.getPropertyValue())
                : DEFAULT_ZOOKEEPER_ADDRESS;
        if (zkClient != null && zkClient.getState() == CuratorFrameworkState.STARTED) {
            return zkClient;
        }
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        zkClient = CuratorFrameworkFactory.builder()
                .connectString(zookeeperAddress)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        try {
            if (!zkClient.blockUntilConnected(30, TimeUnit.SECONDS)) {
                throw new RuntimeException("Time out waiting to connect to ZK!");
            }
        } catch (InterruptedException e) {
            log.error("zkClient error: {}", e.getMessage());
        }
        return zkClient;
    }
}
